<?php

abstract class BaseActivity {
	protected $activity;
	
	protected $cmd;
	protected $env;
	protected $status;
	protected $statusPath;
	protected $resultPath;
	
	protected $taskService;
	protected $resourceService;
	protected $schedulerService;
	
	protected $inputs;
	protected $outputs;
	protected $hdfsSystem;

	public function __construct($activity) {
		$this->activity = $activity;
		
		$this->env = array(		
			'runtime' => getenv('RUNTIME'),
			'taskId' => getenv('TASKID'),
			'jobId' => getenv('JOBID'),
			'workflowName' => getenv('WORKFLOWNAME'),
			'baseDate'	=>	getenv('BASEDATE'),
			'frequency'	=>	getenv('FREQUENCY'),
		);
		$this->statusPath = $this->env['runtime'] . "/log/{$this->activity['activity_name']}.log";
        $this->resultPath = $this->env['runtime'] . "/output/{$this->activity['activity_name']}.out";
        
        if(!is_array($this->activity['inputs'])) {
        	$this->inputs = CJSON::decode($this->activity['inputs']);
        }
        else {
        	$this->inputs = $this->activity['inputs'];
        }
        
        if(!is_array($this->activity['outputs'])) {
			$this->outputs = CJSON::decode($this->activity['outputs']); 
        }
        else {
        	$this->outputs = $this->activity['outputs']; 
        }
			
		$this->status = array();
        $this->taskService = TaskService::getInstance();
        $this->resourceService = ResourceService::getInstance(); 
        $this->schedulerService = SchedulerService::getInstance();
        $this->hdfsSystem = $this->resourceService->getStorageSystemByType(PhysicalType::Hdfs); 
	}
	
	public function execute() {
		$this->taskService->editTaskAttributes($this->env['taskId'], array(
			'detail'	=>	"activity [{$this->activity['activity_name']}] is running",
		));
		$status = $this->prepare();
		if($status == false) {
			Yii::log("activity[{$this->activity['activity_name']}] prepared failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
	
		$status = $this->checkDepends();
		if($status == false) {
			Yii::log("activity[{$this->activity['activity_name']}] depends checked failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
		
		$status = $this->submit();
		if($status == false) {
			Yii::log("activity[{$this->activity['activity_name']}] submitted failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
		
		$status = $this->checkStatus();
		if($status == false) {
			Yii::log("activity[{$this->activity['activity_name']}] status checked failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
		
		$status = $this->save();
		if($status == false) {
			Yii::log("activity[{$this->activity['activity_name']}] results saved failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
		return $status;
	}
	
	protected function prepare() {
		if($this->hasKillCmd()) {
			$this->setRuntimeInfo(array(
				'state'	=>	TaskState::Killed,
				'message'	=>	"task was killed before activity[{$this->activity['activity_name']}] preparing",
			));
			Yii::log("activity[{$this->activity['activity_name']}] is killed by user", CLogger::LEVEL_INFO, WORKER_LOG);
			return false;
		}
		Yii::log("activity[{$this->activity['activity_name']}] is preparing", CLogger::LEVEL_INFO, WORKER_LOG);
		if(!$this->prepareInputs()) {
    		Yii::log("action[{$this->activity['activity_name']}] prepare inputs failed", CLogger::LEVEL_ERROR, WORKER_LOG);
    		return false;
    	}
		if(!$this->prepareOutputs()) {
			Yii::log("action[{$this->activity['activity_name']}] prepare outputs failed", CLogger::LEVEL_ERROR, WORKER_LOG);
    		return false;
    	}
    	
    	$programs = $this->preparePrograms();
    	if($programs === false) {
			Yii::log("action[{$this->activity['activity_name']}] prepare programs failed", CLogger::LEVEL_ERROR, WORKER_LOG);
    		return false;
    	}
    	$this->cmd = $programs[0];
		return true;
	}
	
	protected function checkDepends() {
		if($this->hasKillCmd()) {
			$this->setRuntimeInfo(array(
				'state'	=>	TaskState::Killed,
				'message'	=>	"task was killed before activity[{$this->activity['activity_name']}] checking depends",
			));
			Yii::log("activity[{$this->activity['activity_name']}] is killed by user", CLogger::LEVEL_INFO, WORKER_LOG);
			return false;
		}
		Yii::log("activity[{$this->activity['activity_name']}] is checking depends", CLogger::LEVEL_INFO, WORKER_LOG);
		return true;
	}
	
	protected function submit() {
		if($this->hasKillCmd()) {
			$this->setRuntimeInfo(array(
				'state'	=>	TaskState::Killed,
				'message'	=>	"task was killed before activity[{$this->activity['activity_name']}] submitting",
			));
			Yii::log("activity[{$this->activity['activity_name']}] is killed by user", CLogger::LEVEL_INFO, WORKER_LOG);
			return false;
		}
		
		Yii::log("activity[{$this->activity['activity_name']}] is submitting", CLogger::LEVEL_INFO, WORKER_LOG);
		for($i = 0; $i < $this->activity['retry_times']; $i++) {
            Yii::log('execute cmd: '. $this->cmd, CLogger::LEVEL_INFO, WORKER_LOG);
            exec($this->cmd, $out, $retVal);

            if(!in_array($retVal, array(0))) {
                Yii::log("execute command error[$retVal], retry...", CLogger::LEVEL_ERROR, WORKER_LOG);
                sleep(5*($i + 1));
            }
            else {
                break;
            }
        }
        
        if($retVal != 0) {
        	$this->setRuntimeInfo(array(
				'state'	=>	TaskState::Failed,
				'message'	=>	"execute command for activity[{$this->activity['activity_name']}] failed with[$retVal]",
			));
        }
        return $retVal == 0 ? true : false;
	}
	
 	protected function prepareInputs() {
    	if(is_array($this->inputs)) {
			foreach ($this->inputs as &$input) {			
				$physicalType = $this->getPhysicalType($input['input_type']);
				if($physicalType === false) {
					return false;
				}
				
				$physicalResource = $this->resourceService->getResourcePhysicalByType($input['resource']['res_id'], $physicalType);
		        if(!isset($physicalResource)) {
		            Yii::log("can't find physical as hdfs for resource[{$input['resource']['res_name']}]", CLogger::LEVEL_ERROR, WORKER_LOG);
		            $this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"can't find physical as hdfs for resource[{$input['resource']['res_name']}]",
					));
		            return false;
		        }
		        if($physicalType == PhysicalType::Hive) {
		        	$hiveSystem = $this->resourceService->getStorageSystem($physicalResource['system_id']);
		        	$physicalResource['storage'] = $hiveSystem['namespace'] . '/' . $physicalResource['storage'];
		        }
		        elseif ($physicalType == PhysicalType::Hdfs) {
		        	if($input['input_type'] == ResourceType::ComputeOutput) {
		        		$physicalResource['storage'] = HADOOP_OUTPUT_PATH . '/' . $physicalResource['storage'];
		        	}
		        	elseif(!StringUtil::startsWith($physicalResource['storage'], '/')){
		        		$physicalResource['storage'] = HADOOP_INPUT_PATH . '/' . $physicalResource['storage'];
		        	}
		        }
		        $input['resource']['physical'] = $physicalResource;
		        
		        $exp = DateHelper::exp2partition($input['partition']);
				$input['partitions'] = DateHelper::parseDatePartition($exp['partitions'],
					$this->env['baseDate'],
					$input['resource']['res_frequency'],
					$input['resource']['physical']['partition_format']
				);
			}
    	}
        return true;
    }
    
	protected function prepareOutputs() {
		// Create output physical resouce as HDFS
		if(is_array($this->outputs)) {
			foreach ($this->outputs as &$output) {
				if($output['resource']['res_frequency'] != $this->env['frequency']) {
					Yii::log("resource[{$output['resource']['res_name']}] frequency is not equal to job's frequency", CLogger::LEVEL_ERROR, WORKER_LOG);
		            $this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"resource[{$output['resource']['res_name']}] frequency is not equal to job's frequency",
					));
		            return false;
				}
				
				if(false === $this->resourceService->editResourceAttributes($output['resource']['res_id'], array('job_id'=>$this->env['JOBID']))) {
					$this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"edit output resource [{$output['resource']['res_name']}] job_id failed",
					));
					return false;
				}
								
				$outputPhysical = array(
					'physical_name'	=>	"{$output['resource']['res_name']}_hdfs",
					'res_id'		=>	$output['resource']['res_id'],
					'storage'		=>	$output['resource']['res_name'],
			        'physical_type'	=>	PhysicalType::Hdfs,
					'partition_format'=> 'Ymd/Hi',
			        'system_id'		=>	$this->hdfsSystem['system_id'],
			        'keep_days'		=>	5,
		        );
		        
		        $physicalId = $this->resourceService->addResourcePhysical($outputPhysical);
			 	if(!$physicalId) {
		            Yii::log("add output physical as hdfs for resource[{$output['resource']['res_name']}] failed", CLogger::LEVEL_ERROR, WORKER_LOG);
		            $this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"add output physical as hdfs for resource[{$output['resource']['res_name']}] failed",
					));
		            return false;
		        }
		        $outputPhysical['physical_id'] = $physicalId;
				
				
				$outputPhysical['storage'] = HADOOP_OUTPUT_PATH . "/{$outputPhysical['storage']}";
				$output['resource']['physical'] = $outputPhysical;
				
				$output['partitions'] = DateHelper::parseDatePartition("{0, {$this->env['frequency']}}",
					$this->env['baseDate'],
					$output['resource']['res_frequency'],
					$output['resource']['physical']['partition_format']
				);
			}
		}
		return true;
	}
	
	protected function save() {		
		return true;
	}
	
	protected abstract function checkStatus();
	
	protected abstract function preparePrograms();
	
	protected function hasKillCmd() {
        $conds = array(
            'condition' => array(
                array('task_id', '=', $this->env['taskId']),
                array('state', '=',  CmdState::Submit),
                array('cmd_type', '=',    CommandType::KillTask),
            ),
            'limit'     => 1,
            'order'     => 'cmd_id desc',
        );
        $res = $this->schedulerService->getCommands($conds);
        if($res === false) {
            Yii::log("failed to retrieve kill commands for task: ". $this->env['taskId'], CLogger::LEVEL_ERROR, WORKER_LOG);
            return false;
        }
        if(count($res) == 0) {
            return false;
        }
        
        $row = array(
			'state' => CmdState::Done,
			'accept_time' => DateHelper::makeExactDate(),
        );
        while($this->schedulerService->editCommandAttributes($cmd_id, $row) === false) {
			Yii::log("failed to update command state, keep retrying: ".$cmd_id." row(". print_r($row, true) .")", CLogger::LEVEL_WARNING, WORKER_LOG);
			sleep(2);
		}
		Yii::log("update command state success: ".$cmd_id." row(". print_r($row, true) .")", CLogger::LEVEL_INFO, WORKER_LOG);
        return true;
	}
	
	protected function killTask() {
        $cmd_id = intval($res[0]['cmd_id']);
        $param = CJSON::decode($res[0]['cmd_param']);	        
        $ret = $this->ComputeWrapper->killTask();
        if($ret) {
        	Yii::log('kill task successfully', CLogger::LEVEL_WARNING, WORKER_LOG);
        	$row = array(
				'state' => CmdState::Done,
				'accept_time' => DateHelper::makeExactDate(),
        	);
        }
        else {
            Yii::log('failed to kill task', CLogger::LEVEL_WARNING, WORKER_LOG);
            $row = array(
				'state' => CmdState::Failed,
				'accept_time' => DateHelper::makeExactDate(),
        	);
        }
        
		while($this->schedulerService->editCommandAttributes($cmd_id, $row) === false) {
			Yii::log("failed to update command state, keep retrying: ".$cmd_id." row(". print_r($row, true) .")", CLogger::LEVEL_WARNING, WORKER_LOG);
			sleep(2);
		}
		Yii::log("update command state success: ".$cmd_id." row(". print_r($row, true) .")", CLogger::LEVEL_INFO, WORKER_LOG);
		return $ret;        
	}
	
	protected function getResultSize() {
		return filesize($this->resultPath);
    }
    
    protected function getResultCount() {
		return count(file($this->resultPath));
    }

	protected function setRuntimeInfo($status) {
		$this->status = $status;
    }
    
    public function getRuntimeInfo() {
    	return $this->status;
    }
	
	protected function concreteProgram($program, $params) {
    	foreach($params as $key => $value) {
            $program = str_replace($key, $value, $program);
        }
		return $program;
    } 
    
 	protected function generateColumns($select, $storageType='hive') {    	
    	$columnDefine = '';
    	foreach ($select as $columnName => $columnType) {
    		$type = $this->getColumnType($columnType, $storageType);
    		$columnDefine .= "$columnName $type,";
    	}
    	
    	return substr($columnDefine, 0, strlen($columnDefine) - 1);
    }
    
    protected function generateSelect($select) {
	    $columns = array_keys($select);
	    return implode(',', $columns);
	}
	
	protected function getPhysicalType($inputType) {
    	if($inputType == ResourceType::Log) {
    		$physicalType = PhysicalType::Hdfs;
    	}
    	elseif($inputType == ResourceType::QueryOutput) {
    		$physicalType = PhysicalType::Hive;
    	}
    	elseif($inputType == ResourceType::ComputeOutput) {
    		$physicalType = PhysicalType::Hdfs;
    	}
    	elseif($inputType == ResourceType::RelyFile) {
    		$physicalType = PhysicalType::Hdfs;
    	}
    	else {
    		Yii::log("input type[$inputType] is not support", CLogger::LEVEL_ERROR, WORKER_LOG);
    		return false;
    	}
    	return $physicalType;
    }
	
	private function getColumnType($type, $storageType) {
    	if($storageType == 'hive') {
    		return $type['type'];
    	}
    	else {
    		if($type['type'] == 'int') {
    			return 'int(11) unsigned';
    		}
    		elseif($type['type'] == 'string') {
    			return "varchar({$type['length']}) COLLATE utf8_bin";
    		}
    	}
    }
}
